package com.service;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Timer;
import java.util.TimerTask;

@Service
public class RabbitMQService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    long[] times = {1000, 1000 * 10, 10000 * 100, 1000 * 1000, 1000 * 10000};

    /**
     * 发送消息
     *
     * @param msg
     */
    public void sendMessage(String exchcange, String key, Message msg) {
        //交换机不存在的反馈
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if (b == false) {
                    System.out.println("交换机不存在!");
                }
            }
        });

        //key不存在时反馈
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                System.out.println("key搞错答");
            }
        });


        rabbitTemplate.convertAndSend(exchcange, key, msg);
    }


    /**
     * 监听消息
     *
     * @param msg
     * @throws UnsupportedEncodingException
     */
    @RabbitListener(queues = "s2020_2_3_queue")
    @RabbitHandler
    public void recevieMQ(Message msg,final Channel channel) throws Exception {
        try {
            int n = 1 / 0;

            //已经正常处理好消息了: 消息编号/是否批量处理所有比当前编号小的消息
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);

        } catch (Exception e) {
            //是否重发消息,一般不用下面这个，会死循环，后面消息也无法读取
            // channel.basicReject(msg.getMessageProperties().getDeliveryTag(),true);
            String uuid = msg.getMessageProperties().getMessageId();
            Integer count = (Integer) redisTemplate.boundValueOps(uuid).get();
            if (count == null) {
                count = 0;
            }
            count++;
            final int s = count;
            if (count < 5) {
                System.out.println("消息编号:"+msg.getMessageProperties().getDeliveryTag());
                String str = new String(msg.getBody(),"utf-8");
                System.out.println("收到消息:"+str);

                new Timer().schedule(new TimerTask() {
                      @Override
                      public void run() {

                          redisTemplate.boundValueOps(uuid).set(s);
                          //丢弃消息: 消息编号，是否批量，是否重发,一般不用true,重试次要自己处理好
                          try {
                              channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, false);
                              //重发
                              rabbitTemplate.convertAndSend(msg.getMessageProperties().getReceivedExchange(), msg.getMessageProperties().getReceivedRoutingKey(), msg);

                          } catch (IOException e1) {
                              e1.printStackTrace();
                          }

                      }
                  },times[count-1]);
            } else {
                // System.out.println("另外处理");
                return;
            }
        }


    }

}










